package rabbit

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/xinshortlink/internal/pkg/conf"
	"gitee.com/xinshortlink/internal/pkg/mq"
	"github.com/avast/retry-go/v4"
	"github.com/golang/snappy"
	"github.com/rabbitmq/amqp091-go"
	"go.uber.org/zap"
	"golang.org/x/sync/semaphore"
	"sync"
	"sync/atomic"
	"time"
)

const (
	deadExchange   = "dead_exchange"    // 死信交换机
	deadQueue      = "dead_queue"       // 死信队列
	deadRoutingKey = "dead_routing_key" // 死信路由键
	retryCountKey  = "retry_count"
)

type Rabbit struct {
	sync.Mutex

	available atomic.Bool
	conn      *amqp091.Connection
	ch        *amqp091.Channel

	dsn           string
	exchangeName  string
	routingKey    string
	queueName     string
	prefetchCount int
	callback      mq.ConsumeFunc
	sem           *semaphore.Weighted
	logger        *zap.Logger
}

func NewRabbit(conf *conf.Conf, logger *zap.Logger) (*Rabbit, error) {
	r := &Rabbit{
		logger:        logger,
		dsn:           conf.Mq.Rabbit.Dsn,
		routingKey:    conf.Mq.Rabbit.RoutingKey,
		exchangeName:  conf.Mq.Rabbit.ExchangeName,
		queueName:     conf.Mq.Rabbit.QueueName,
		prefetchCount: conf.Mq.Rabbit.PrefetchCount,
		sem:           semaphore.NewWeighted(int64(conf.Mq.Rabbit.MaxConcurrent)),
	}

	r.available.Store(false)

	err := r.dial()
	if err != nil {
		return nil, err
	}

	err = r.channel()
	if err != nil {
		return nil, err
	}

	r.available.Store(true)

	go r.monitorConn()
	go r.monitorChan()
	go r.consumeMessages()

	return r, err
}

// dial 连接
func (r *Rabbit) dial() error {
	err := retry.Do(func() error {
		r.Lock()
		defer r.Unlock()

		conn, err := amqp091.DialConfig(r.dsn, amqp091.Config{
			Heartbeat: 10 * time.Second,
			Dial:      amqp091.DefaultDial(20 * time.Second),
		})
		if err != nil {
			return err
		}

		r.conn = conn
		return nil
	}, retry.Attempts(5), retry.DelayType(retry.BackOffDelay), retry.OnRetry(func(n uint, err error) {
		r.logger.Error("重新发起链接")
	}))

	return err
}

// monitorConn 监控连接
func (r *Rabbit) monitorConn() {
	// 启动协程监听是否断线

	closeCh := make(chan *amqp091.Error)
	r.conn.NotifyClose(closeCh)

	for {
		select {
		case err, ok := <-closeCh:
			if !ok {
				r.logger.Info("主动关闭链接")
				return
			}

			if err != nil {
				r.logger.Error("rabbit mq 链接关闭", zap.Error(err))

				err := r.dial()

				if err != nil {
					r.logger.Error("重试链接失败, rabbitmq 不可用", zap.Error(err))

					r.available.Store(false) // 设置为不可用
				}
			}
		}
	}
}

// channel channel
func (r *Rabbit) channel() error {
	err := retry.Do(func() error {
		r.Lock()
		defer r.Unlock()

		ch, err := r.conn.Channel()
		if err != nil {
			return err
		}

		// PrefetchCount 单个消费者最多能同时处理的未确认消息个数
		err = ch.Qos(r.prefetchCount, 0, false)
		if err != nil {
			return err
		}

		// 声明死信交换机
		err = ch.ExchangeDeclare(deadExchange, amqp091.ExchangeDirect, true, false, false, false, nil)
		if err != nil {
			return err
		}

		// 声明死信队列
		_, err = ch.QueueDeclare(deadQueue, true, false, false, false, nil)
		if err != nil {
			return err
		}

		// 死信队列绑定
		err = ch.QueueBind(deadQueue, deadRoutingKey, deadExchange, false, nil)
		if err != nil {
			return err
		}

		// 声明主交换机
		err = ch.ExchangeDeclare(r.exchangeName, amqp091.ExchangeDirect, true, false, false, false, nil)
		if err != nil {
			return err
		}

		args := amqp091.Table{
			"x-dead-letter-exchange":    deadExchange,   // 死信交换机名
			"x-dead-letter-routing-key": deadRoutingKey, // 路由键
		}

		// 声明主队列
		_, err = ch.QueueDeclare(r.queueName, true, false, false, false, args)
		if err != nil {
			return err
		}

		// 主队列绑定
		err = ch.QueueBind(r.queueName, r.routingKey, r.exchangeName, false, nil)
		if err != nil {
			return err
		}

		if err := ch.Confirm(false); err != nil { // 开启确认模式
			return err
		}

		r.ch = ch

		return nil
	}, retry.Attempts(5), retry.DelayType(retry.BackOffDelay), retry.OnRetry(func(n uint, err error) {
		r.logger.Error("重新发起链接")
	}))

	return err
}

// monitorChan 监控通道
func (r *Rabbit) monitorChan() {
	closeCh := make(chan *amqp091.Error)
	r.ch.NotifyClose(closeCh)

	for {
		select {
		case err, ok := <-closeCh:
			if !ok {
				r.logger.Info("主动关闭通道")
				return
			}

			if err != nil {
				r.logger.Error("rabbit mq 通道关闭", zap.Error(err))

				err := r.channel()

				if err != nil {
					r.logger.Error("重试创建通道失败, rabbitmq 不可用", zap.Error(err))

					r.available.Store(false) // 设置为不可用
				}
			}
		}
	}
}

func (r *Rabbit) Register(callback mq.ConsumeFunc) {
	r.Lock()
	defer r.Unlock()

	r.callback = callback // 设置消费消息函数
}

func (r *Rabbit) Shutdown() {
	r.Lock()
	defer r.Unlock()

	r.available.Store(false)

	if r.conn != nil {
		_ = r.conn.Close()
	}

	if r.ch != nil {
		_ = r.ch.Close()
	}
}

var h = amqp091.Table{
	retryCountKey: 0,
}

func (r *Rabbit) send(message []byte, headers amqp091.Table) error {
	// 发送前增加通道检测逻辑
	if r.ch.IsClosed() {
		if err := r.channel(); err != nil {
			return fmt.Errorf("channel recovery failed: %v", err)
		}
	}

	bytes := snappy.Encode(nil, message) // 压缩

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	confirm, err := r.ch.PublishWithDeferredConfirmWithContext(ctx, r.exchangeName, r.routingKey, true, false, amqp091.Publishing{
		DeliveryMode: amqp091.Persistent, // 持久化
		ContentType:  "application/protobuf",
		Body:         bytes,
		Headers:      headers,
	})

	if err != nil {
		return err
	}

	if !confirm.Wait() {
		return fmt.Errorf("消息未确认 (confirmed=%v, err=%v)", false, err)
	}

	return nil
}

func (r *Rabbit) Send(message []byte) error {
	if !r.available.Load() {
		return errors.New("rabbit mq 不可用")
	}

	return r.send(message, h)
}

func (r *Rabbit) processMessage(d amqp091.Delivery) {
	defer r.sem.Release(1) // 释放信号量

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	// 开启一个协程消费消息
	done := make(chan error, 1)
	go func(msg []byte) {
		defer func() {
			if r := recover(); r != nil {
				done <- fmt.Errorf("process message panic: %v", r)
			}
		}()

		bytes, err := snappy.Decode(nil, msg)
		if err != nil {
			done <- err
			return
		}

		done <- r.callback(ctx, bytes) // 消费消息
	}(d.Body)

	select {
	case <-ctx.Done():
		// 超时消息
		r.nack(d)
	case err := <-done:
		if err != nil {
			r.nack(d)
		} else {
			// 确认消息
			r.ack(d)
		}
	}
}

func (r *Rabbit) consumeMessages() {
	// 消费返回的为一个无缓冲的通道
	delivery, err := r.ch.Consume(r.queueName, "", false, false, false, false, nil)
	if err != nil {
		r.logger.Error("启动消费者失败", zap.Error(err))
		return
	}

	// for 循环阻塞消费
	for d := range delivery {
		if r.callback == nil {
			continue
		}
		if err := r.sem.Acquire(context.Background(), 1); err != nil {
			r.logger.Error("获取信号量失败", zap.Error(err))
			continue
		}

		// 此处开启协程的作用是使用 context 控制消息消费时间不能超过3秒
		go r.processMessage(d)
	}
}

// ack 确认消息
// amqp091.Delivery 为值传递。可以考虑优化为指针。
func (r *Rabbit) ack(d amqp091.Delivery) error {
	if err := d.Ack(false); err != nil {
		r.logger.Error("确认消息失败", zap.String("messageId", d.MessageId), zap.Error(err))
		return err
	}
	return nil
}

// nack 将消息返回消息队列
// 重试3次。失败后放入死信队列
func (r *Rabbit) nack(d amqp091.Delivery) {
	retryCount, _ := d.Headers[retryCountKey].(int32)

	r.logger.Info(">>>retryCount", zap.Int32("retryCount", retryCount))

	if retryCount >= 3 {
		r.logger.Info(">>> 重试次数太多，进入死信队列", zap.String("messageId", d.MessageId))
		err := d.Nack(false, false) // 超过重试次数，进入死信队列

		if err != nil {
			r.logger.Error("reject error", zap.String("messageId", d.MessageId), zap.Error(err))
		}

	} else {
		headers := amqp091.Table{
			retryCountKey: retryCount + 1,
		}

		if err := r.send(d.Body, headers); err == nil {
			err = r.ack(d)

			if err != nil {
				r.logger.Error("确认消息失败,直接进入死信队列", zap.String("messageId", d.MessageId), zap.Error(err))
				err := d.Nack(false, false) // 超过重试次数，进入死信队列
				if err != nil {
					r.logger.Error("reject error", zap.String("messageId", d.MessageId), zap.Error(err))
				}
				return
			}
		} else {
			r.logger.Error("重试消息失败", zap.Error(err))
		}
	}

}
